package ua.com.softway.db.batch;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

import javax.persistence.EntityManager;
import javax.persistence.EntityManagerFactory;
import javax.persistence.EntityTransaction;

import org.apache.log4j.Logger;
import org.hibernate.Criteria;
import org.hibernate.ScrollMode;
import org.hibernate.ScrollableResults;

import ua.com.softway.db.Persistence;

public class ScrollableBatchProcessor implements Runnable {
    private final static Logger log = Logger.getLogger(ScrollableBatchProcessor.class);

    @SuppressWarnings("unchecked")
    private void process(BatchJob job) {
        BatchFailed batchFailed = job.getClass().getAnnotation(BatchFailed.class);

        if (batchFailed != null && batchFailed.value() != BatchFailedPolicy.Abort)
            throw new IllegalArgumentException("ScrollableBatchProcessor do not supports BatchFailedPolicies others then Abort");
        
        int batchSize = job.getBatchSize();
        
        EntityManagerFactory entityManagerFactory = Persistence.getEntityManagerFactory();
        
        EntityManager entityManager = entityManagerFactory.createEntityManager();
        Persistence.setCurrentEntityManager(entityManager);
        
        EntityTransaction transaction = null;
        
        try {
            transaction = entityManager.getTransaction();
            transaction.begin();
        
            Criteria query = job.createCriteria();
            query.setFetchSize(batchSize);
            
            ScrollableResults scroll = query.scroll(ScrollMode.FORWARD_ONLY);
            
            while (true) {
                List resultList = getNextResults(scroll, batchSize);
                int size = resultList.size();
                
                if (size == 0) break;
                
                job.processBatch(resultList);
                
                entityManager.flush();
                entityManager.clear();
                
                if (size < batchSize) break;
            }
            
            job.completed();
            
            if (transaction.isActive())
                transaction.commit();
        } catch(RuntimeException e) {
            log.error("Can't complete job " + currentJob, e);
            
            job.failed();
            
            throw e;
        } finally {
            if ( transaction != null && transaction.isActive() ) transaction.rollback();

            entityManager.close();
            Persistence.setCurrentEntityManager(null);
        }
    }
    
    @SuppressWarnings("unchecked")
    private List getNextResults(ScrollableResults scroll, int batchSize) {
        List r = new ArrayList();
        
        while (batchSize -- > 0) {
            if (!scroll.next())
                break;
            
            r.add(scroll.get(0));
        }
        
        return r;
    }

    final BlockingQueue<BatchJob<?>> jobs;
    
    public void addJob(BatchJob<?> job) {
        jobs.add(job);
    }
    
    BatchJob<?> currentJob = null;
    
    @Override
    public void run() {
        while (true) {
            try {
                currentJob = jobs.take();
                
                try {
                    process(currentJob);
                } catch(RuntimeException e) {
                } finally {
                    currentJob = null;
                    
                    synchronized (this) {
                        this.notifyAll();
                    }
                }
            } catch (InterruptedException e) {
                return;
            }
        }
    }

    public void waitForCompletion() throws InterruptedException {
        while (isProcessing()) {
            synchronized (this) {
                wait();
            }
        }
    }

    public boolean isProcessing() {
        return currentJob != null || !jobs.isEmpty();
    }
    
    ScrollableBatchProcessor(int queueSize) {
        jobs = new ArrayBlockingQueue<BatchJob<?>>(queueSize);
    }
    
    private static ScrollableBatchProcessor instance;
    
    public static ScrollableBatchProcessor init(int queueSize) {
        if (instance != null) throw new IllegalStateException();
        
        instance = new ScrollableBatchProcessor(queueSize);
        
        return instance;
    }
    
    public static ScrollableBatchProcessor init() {
        return init(30);
    }
    
    public static ScrollableBatchProcessor instance() {
        if (instance == null) throw new IllegalStateException();
        
        return instance;
    }
}
